package com.lixixi.tingshu.album.listener;

import com.alibaba.fastjson.JSONObject;
import com.lixixi.tingshu.album.mapper.TrackStatMapper;
import com.lixixi.tingshu.common.constant.KafkaConstant;
import com.lixixi.tingshu.common.constant.RedisConstant;
import com.lixixi.tingshu.vo.album.TrackStatMqVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class TrackStatUpdateListener {
	@Autowired
	RedisTemplate redisTemplate;

	@Autowired
	TrackStatMapper trackStatMapper;

	@KafkaListener(id = "updateTrackStat", topics = KafkaConstant.QUEUE_TRACK_STAT_UPDATE)
	void trackStatListener(String trackStatMqVoJson) {
		TrackStatMqVo trackStatMqVo = JSONObject.parseObject(trackStatMqVoJson, TrackStatMqVo.class);
		//使用redis实现分布式锁,标记该消息是否已消费
		//维持10分钟
		Boolean absent = redisTemplate.opsForValue().setIfAbsent(trackStatMqVo.getBusinessNo(), 1, RedisConstant.CACHE_TEMPORARY_TIMEOUT, TimeUnit.SECONDS);
		if (Boolean.FALSE.equals(absent))
		{
			//标识该消息已被消费,不做处理
			return;
		}
		try
		{
			Assert.notNull(trackStatMqVo, "消息传递失败");
			log.info("消息编号:{}-专辑:{}-声音:{}-统计类型:{}-更新:{}"
					, trackStatMqVo.getBusinessNo()
					, trackStatMqVo.getAlbumId()
					, trackStatMqVo.getTrackId()
					, trackStatMqVo.getStatType()
					, trackStatMqVo.getCount());
			trackStatMapper.updateTrackStat(trackStatMqVo.getStatType(), trackStatMqVo.getCount(), trackStatMqVo.getTrackId());
		} catch (IllegalArgumentException exception)
		{
			//todo:对该消息进行特殊处理
			log.error("消息消费失败:{}", trackStatMqVoJson);
		}
	}
}
